#include <ddm/OpenrAgent.h>
#include <folly/logging/xlog.h>
#include <openr/common/NetworkUtil.h>
#include <openr/kvstore/KvStoreUtil.h>

#include <iostream>

namespace ddm {
OpenrAgent::OpenrAgent(
    std::string nodeName, std::shared_ptr<ddm::Config> config,
    openr::messaging::ReplicateQueue<ddm::thrift::Requirement>&
        requirementsQueue)
    : nodeName_(nodeName), config_(config), requirementsQueue_(requirementsQueue) {
  openrClient_ =
      openr::getOpenrCtrlPlainTextClient<openr::thrift::OpenrCtrlCppAsyncClient,
                                         apache::thrift::RocketClientChannel>(
          *this->getEvb(), folly::IPAddress("::1"), 2018,
          std::chrono::milliseconds(2000), std::chrono::milliseconds(0));
}

void OpenrAgent::stop() { openr::OpenrEventBase::stop(); }

void OpenrAgent::subscribeNeighbor() {
  openr::thrift::KeyDumpParams params;
  std::string key = "dst:" + nodeName_ + ".*";
  XLOG(INFO) << "Start subscribing key: " << key;
  params.keys() = {key};
  auto response =
      openrClient_->semifuture_subscribeAndGetAreaKvStores(params, {"0"}).get();
  std::unordered_map<
      std::string /* area */,
      std::unordered_map<std::string /* key */, openr::thrift::Value>>
      areaKeyVals;
  XLOG(INFO) << "Stream is connected, updates will follow";
  for (auto const& pub : response.response) {
    XLOG(INFO) << "Received " << pub.keyVals()->size()
               << " entries in initial dump for area: " << *pub.area();
    areaKeyVals[*pub.area()] = *pub.keyVals();
  }
  XLOG(INFO) << "";

  auto subscription =
      std::move(response.stream)
          .subscribeExTry(
              folly::Executor::getKeepAliveToken(this->getEvb()),
              [areaKeyVals = std::move(areaKeyVals), this](
                  folly::Try<openr::thrift::Publication>&& maybePub) mutable {
                XLOG(INFO) << "Got kv update";
                if (maybePub.hasException()) {
                  XLOG(ERR) << maybePub.exception().what();
                  return;
                }
                auto& pub = maybePub.value();
                // Print expired key-vals
                for (const auto& key : *pub.expiredKeys()) {
                  std::cout << "Expired Key: " << key << std::endl;
                  std::cout << "" << std::endl;
                }

                // Print updates
                auto updatedKeyVals =
                    openr::mergeKeyValues(areaKeyVals.at(pub.get_area()),
                                          *pub.keyVals())
                        .first;
                for (auto& [key, val] : updatedKeyVals) {
                  std::cout
                      << (val.value().has_value() ? "Updated" : "Refreshed")
                      << " KeyVal: " << key << std::endl;
                  std::cout << "  version: " << *val.version() << std::endl;
                  std::cout << "  originatorId: " << *val.originatorId()
                            << std::endl;
                  std::cout << "  ttl: " << *val.ttl() << std::endl;
                  std::cout << "  ttlVersion: " << *val.ttlVersion()
                            << std::endl;
                  std::cout << "  hash: " << val.hash().value() << std::endl
                            << std::endl;  // intended
                }
              });
  std::move(subscription).detach();
}

void OpenrAgent::startSubscribe() {
  // Query neighbors and set listening keys
  auto resp = openrClient_->semifuture_getNeighbors().get();
  std::map<std::string, std::string> ifaceToNeighborName;
  for (auto const& pub : resp) {
    XLOG(INFO) << "Get neighbor: " << *pub.nodeName() << *pub.remoteIfName();
    ifaceToNeighborName[*pub.localIfName()] = *pub.nodeName();
  }
  // config_->setIfaceToNeighborName(ifaceToNeighborName);

  // subscribeNeighbor();
  openr::thrift::KeyDumpParams params;
  std::string toMeKey = "dst:" + nodeName_ + ".*";
  std::string reqKey = "req:.*";
  XLOG(INFO) << "Start subscribing key: " << toMeKey << " " << reqKey;
  params.keys() = {toMeKey, reqKey};

  auto response =
      openrClient_->semifuture_subscribeAndGetAreaKvStores(params, {"0"}).get();
  std::unordered_map<
      std::string /* area */,
      std::unordered_map<std::string /* key */, openr::thrift::Value>>
      areaKeyVals;
  XLOG(INFO) << "Stream is connected, updates will follow";
  for (auto const& pub : response.response) {
    XLOG(INFO) << "Received " << pub.keyVals()->size()
               << " entries in initial dump for area: " << *pub.area();
    areaKeyVals[*pub.area()] = *pub.keyVals();
  }
  XLOG(INFO) << "";

  openrClient_->semifuture_getMyNodeName();

  // std::string nodeName_ = "A";
  // openr::thrift::KeyVals kvs(
  //     {{"key1", openr::createThriftValue(1, nodeName_, std::string("value1"))},
  //      {"key2", openr::createThriftValue(1, nodeName_, std::string("value2"))},
  //      {"key3",
  //       openr::createThriftValue(1, nodeName_, std::string("value3"))}});
  // openr::thrift::KeySetParams params;
  // params.keyVals() = kvs;
  // XLOG(INFO) << "Send update";
  // openrClient_->semifuture_setKvStoreKeyVals(params, "0");

  auto subscription =
      std::move(response.stream)
          .subscribeExTry(
              folly::Executor::getKeepAliveToken(this->getEvb()),
              [areaKeyVals = std::move(areaKeyVals), this](
                  folly::Try<openr::thrift::Publication>&& maybePub) mutable {
                XLOG(INFO) << "Got kv update";
                if (maybePub.hasException()) {
                  XLOG(ERR) << maybePub.exception().what();
                  return;
                }
                auto& pub = maybePub.value();
                // Print expired key-vals
                for (const auto& key : *pub.expiredKeys()) {
                  std::cout << "Expired Key: " << key << std::endl;
                  std::cout << "" << std::endl;
                }

                // Print updates
                auto updatedKeyVals =
                    openr::mergeKeyValues(areaKeyVals.at(pub.get_area()),
                                          *pub.keyVals())
                        .first;
                for (auto& [key, val] : updatedKeyVals) {
                  std::cout
                      << (val.value().has_value() ? "Updated" : "Refreshed")
                      << " KeyVal: " << key << std::endl;
                  std::cout << "  version: " << *val.version() << std::endl;
                  std::cout << "  originatorId: " << *val.originatorId()
                            << std::endl;
                  std::cout << "  ttl: " << *val.ttl() << std::endl;
                  std::cout << "  ttlVersion: " << *val.ttlVersion()
                            << std::endl;
                  std::cout << "  hash: " << val.hash().value() << std::endl
                            << std::endl;  // intended

                  // TODO: use filtered subscription
                  if (key.rfind("req", 0) == 0) {
                    apache::thrift::CompactSerializer serializer;
                    auto req =
                        openr::readThriftObjStr<ddm::thrift::Requirement>(
                            *val.value(), serializer);
                    std::cout << "Got requirement: " << key << " val: "
                              << openr::toIPNetwork(*req.prefix()).first.str()
                              << " " << *req.prefix()->prefixLength()
                              << std::endl;
                    requirementsQueue_.push(req);
                  }
                }
              });
  std::move(subscription).detach();
}
}  // namespace ddm